# import sys
# print(sys.path)
# import sklearn
# import mrjob
# print(mrjob.__file__)
from mrjob.job import MRJob
import re
from mr3px.csvprotocol import CsvProtocol

"""
需要处理的数据
	Category有3类		Sub-Category有20类左右  
	获取每行数据中的Ship Date、Segment、Country、City、State、Region、Category、Sub-Category、Quantity、Discount、Profit
	1.将日期合并，统计同一年内，不同一级商品类型销售的数量和利润
	2.将日期合并，统计同一年内，不同国家销售的数量和利润
	3.将Category合并，统记每年的利润走向
	4.将Sub-Category合并，统计每年的利润走向
	5.将地区和一级分类合并，统计每年地区的利润走向
	6.将地区和二级分类合并，统计每年地区的利润走向
	Row ID,Order ID,Order Date,Ship Date,Ship Mode,Customer ID,Customer Name,Segment,Country,City,State,Postal Code,Region,Product ID,Category,Sub-Category,Product Name,Sales,Quantity,Discount,Profit

	
"""


class MRwordCount(MRJob):
    '''
        line:一行数据						出货日期	运输模式											划分国家		城市		洲						地区					          一级分类	二级分类		产品名称									销售							盈利
        	0	1				2			3			4				5			6						7				8			9			10			11		12				13			14				15		16									17			18			19		20		
        Row ID	Order ID		Order Date	Ship Date	Ship Mode		Customer ID	Customer Name			Segment		Country		City		State		Postal Code	Region	Product ID		Category	Sub-Category	Product Name						Sales	Quantity	Discount	Profit
			1	CA-2017-152156	2017/11/8	2017/11/11	Second Class	CG-12520	Claire Gute	Consumer	United 		States		Henderson	Kentucky	42420		South	FUR-BO-10001798	Furniture	Bookcases		Bush Somerset Collection Bookcase	261.96	2			0			41.9136
		1.将日期合并，统计同一年内，不同一级商品类型销售的数量和利润
		返回结果，例：
			year	category   quantity      profit
			2019	f           10           100
			2019	m           23           203
			2020	f           10           100
			2020	m           23           203
		mapper1:(year, (category, quantity, profit)) 
			
		reducer1:
				shuff and sort 之后:
			例：(2019, [(f, 10, 100), (m, 11, 101), (m, 12, 102)])
				(2020, [(f, 10, 100), (m, 11, 101), (m, 12, 102)])
'''
    # OUTPUT_PROTOCOL = CsvProtocol  # write output as CSV
    INPUT_PROTOCOL = CsvProtocol

    def mapper(self, _, data_list):
        # line是文件中的每一行
        # data_list = line.split(",")
        if len(data_list)<21:
        	return
        # if data_list[0] == b"Row ID":
        #     return
        if re.match(r'^[0-9]', data_list[0]) is None:  # 没报错说明是string类型
        	return
        # print(bytes(data_list))
        year = data_list[2].split("/")[0]
        category = data_list[14]
        quantity = data_list[18]
        profit = data_list[20]
        # print((year, (category, quantity, profit)))
        yield (year, (category, quantity, profit))

        # pattern=re.compile(r'(\W+)')
        # for word in re.split(pattern=pattern,string=line):
        #     if word.isalpha():
        #         yield (word.lower(),1)

    def reducer(self, year, data_list):
        # shuff and sort 之后
        '''
        (2019, [(f, 10, 100), (m, 11, 101), (m, 12, 102)])
        (2020, [(f, 10, 100), (m, 11, 101), (m, 12, 102)])
        '''
        category_list = ["Furniture", "Office Supplies", "Technology"]
        f_quantity = 0
        f_profit = 0
        o_quantity = 0
        o_profit = 0
        t_quantity = 0
        t_profit = 0

        for data in data_list:
            if data[0] == category_list[0]:
                f_quantity += int(data[1])
                f_profit += float(data[2])
            elif data[0] == category_list[1]:
                o_quantity += int(data[1])
                o_profit += float(data[2])
            elif data[0] == category_list[2]:
                t_quantity += int(data[1])
                t_profit += float(data[2])
        res_list = [(category_list[0], f_quantity, f_profit), (category_list[1], o_quantity, o_profit),
                    (category_list[2], t_quantity, t_profit)]
        # print(res_list)
        yield (year, res_list[0])
        yield (year, res_list[1])
        yield (year, res_list[2])
        # l=list(count)
        # yield (word,sum(l))


if __name__ == '__main__':
    # print(sys.path)
    MRwordCount.run()  # run()方法，开始执行MapReduce任务。
